package org.fastsyncer.manager.listener.mysql;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.entity.DatabaseConfig;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.connector.mysql.MysqlConnector;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

public final class SqlMysqlListener extends AbstarctMysqlListener {

    private static final Logger logger = LoggerFactory.getLogger(SqlMysqlListener.class);

	protected MysqlConnector mysqlConnector = MysqlConnector.getInstance();
	
    @Override
    protected String getTable() {
        MappingTask mapping = super.getMapping();
        Map<String, String> inc = mapping.getPolicy().get(MappingConstant.POLICY_INCREMENT);
        String table = inc.get("table");
        // 保证配置表信息正确
        super.config.setTable(table);
        return table;
    }

    @Override
    protected List<Map<String, Object>> getColNames() {
        /**
         * 因为SqlMysql连接器是通过SQL查询指定列信息,无法获取增量表所有的列信息，
         * 所以这里使用Mysql连接器查询表所有的列信息。
         */
        return mysqlConnector.getBaseInfoConnector(super.config).getMetaInfo();
    }

    @Override
    protected void sendMsg(JSONArray msg) {
        try {
            // 使用联合查询获取并集
            msg = this.query(msg);
            // 发送消息
            super.sendMsg(msg);
        } catch (JSONException e) {
            logger.error("Error", e.getLocalizedMessage());
        }
    }

    private JSONArray query(JSONArray msg) throws JSONException {
        MappingTask mapping = super.getMapping();
        Mapping source = mapping.getSourceMapping();
        List<Map<String, Object>> fileds = source.getFileds();
        if (null == fileds || fileds.isEmpty()) {
            return null;
        }
        // 获取需要同步的字段(为了排除非必须合并的关联数据,优化性能)
        List<String> required = this.getFileds(fileds);
        Map<String, String> executeCommond = source.getExecuteCommond();
        // 1.获取主键
        DatabaseConfig config = (DatabaseConfig) source.getConfig();

        // 2.建立连接
        JdbcTemplate jdbcTemplate = mysqlConnector.getJdbcTemplate(config);
        
        // 3.定义一个最终结果集
        JSONArray rs = new JSONArray();
        try {
            JSONObject rowObj;
            JSONArray row;
            String pkValue;
            Map<String, Object> data;
            String pk = config.getPrimaryKey();
            // 4.获取联合SQL + 增量SQL条件
            String sql = config.getSql() + executeCommond.get(ConnectorConstant.OPERTION_QUERY_LOG);
            
            int length = msg.length();
            for (int i = 0; i < length; i++) {
                rowObj = msg.getJSONObject(i);
                // 5.排除删除事件
                if(ConnectorConstant.OPERTION_DELETE.equals(rowObj.getString("eventType"))){
                    rs.put(rowObj);
                    continue;
                }
                // 6.获取增量数据唯一主键
                row = rowObj.getJSONArray("after");
                pkValue = this.getPkValueFromRow(pk, row);
                if (StringUtils.isBlank(pkValue)) {
                    continue;
                }
                // 7.获取关联数据
                data = this.query(jdbcTemplate, sql, new Object[] { pkValue });
                // 如果不存在关联,那么整条数据无意义,跳过
                if(null == data){
                    continue;
                }
                // 8.将关联的数据合并到增量数据中
                rowObj.put("after", new JSONArray());
                row = rowObj.getJSONArray("after");
                this.merge(row, data, required);
                rs.put(rowObj);
            }
            data = null;
            required = null;
        } catch (Exception e) {
            logger.error(e.getClass() + " >> " + e.getLocalizedMessage());
        } finally {
            // 9.断开连接
            mysqlConnector.close(jdbcTemplate);
        }
        return rs;
    }

    private List<String> getFileds(List<Map<String, Object>> fileds) {
        List<String> m = new ArrayList<String>();
        int size = fileds.size();
        Map<String, Object> c;
        String name;
        for (int i = 0; i < size; i++) {
            c = fileds.get(i);
            name = c.get("name").toString();
            // 去空、去重
            if(StringUtils.isBlank(name) || m.contains(name)){
                continue;
            }
            m.add(name);
        }
        return m;
    }

    /**
     * @param row 从binlog抓取的原始数据(空数组)
     * @param data 通过sql查询关联的数据
     * @param required 必需设置的字段
     * @return 合并后的结果集
     * @throws JSONException
     */
    private void merge(JSONArray row, Map<String, Object> data, List<String> required) throws JSONException {
        // row=[]
        // data={"name":"123","value": "3"},{"phone":"138xxx","value": "xxx"}
        // required={"name"}
        // 思路：向row追加关联的必须字段
        String k;
        JSONObject n;
        Object v;
        int size = required.size();
        for (int i = 0; i < size; i++) {
            k = required.get(i);
            v = data.get(k);
            if(null == v){
                continue;
            }
            n = new JSONObject();
            n.put("name", k);
            n.put("value", v);
            row.put(n);
        }
    }

    private String getPkValueFromRow(String pk, JSONArray row) throws JSONException {
        int length = row.length();
        String v = null;
        JSONObject col;
        for (int i = 0; i < length; i++) {
            col = row.getJSONObject(i);
            // 找到主键,返回主键值
            if (pk.equalsIgnoreCase(col.getString("name"))) {
                v = col.getString("value");
                continue;
            }
        }
        return v;
    }

    private Map<String, Object> query(JdbcTemplate jdbcTemplate, String sql, Object[] params) {
        try {
            return jdbcTemplate.queryForMap(sql, params);
        } catch (Exception e) {
            if(logger.isDebugEnabled()){
                logger.debug(e.getClass() + " >> " + e.getLocalizedMessage());
            }
        }
        return null;
    }
    
}
